//! # Publisher Trait 完整单元测试
//!
//! 提供 Publisher trait 和具体实现的全面测试覆盖

#[cfg(test)]
mod publisher_tests {
    use crate::prelude::*;
    use crate::mock::{MockMessageBus, MockBehavior, test_helpers::*};
    use crate::publisher::{SimplePublisher, BatchPublisher};
    use serde_json::json;
    use std::sync::Arc;
    use tokio::time::{timeout, Duration};
    use uuid::Uuid;

    /// 测试 SimplePublisher 的所有场景
    mod simple_publisher_tests {
        use super::*;

        async fn create_simple_publisher() -> (SimplePublisher, Arc<MockMessageBus>) {
            let bus = Arc::new(MockMessageBus::new());
            let publisher = SimplePublisher::new(bus.clone(), "test_publisher".to_string());
            (publisher, bus)
        }

        #[tokio::test]
        async fn test_simple_publisher_creation() {
            let (publisher, _) = create_simple_publisher().await;
            assert_eq!(publisher.id(), "test_publisher");
        }

        #[tokio::test]
        async fn test_publish_single_message() {
            let (publisher, bus) = create_simple_publisher().await;
            
            let message = create_test_message("test.topic", json!({"simple": "publish"}));
            let result = publisher.publish(message.clone()).await;
            
            assert!(result.is_ok());
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
            assert_eq!(published[0].topic(), "test.topic");
            assert_eq!(published[0].payload()["simple"], "publish");
        }

        #[tokio::test]
        async fn test_publish_different_message_types() {
            let (publisher, bus) = create_simple_publisher().await;
            
            // 事件消息
            let event = create_event_message("events", json!({"type": "user_created"}));
            publisher.publish(event).await.unwrap();
            
            // 命令消息
            let command = create_command_message("commands", json!({"action": "delete_user"}));
            publisher.publish(command).await.unwrap();
            
            // 查询消息
            let query = create_request_message("queries", json!({"query": "get_user"}), "reply.topic");
            publisher.publish(query).await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 3);
            assert_eq!(published[0].metadata.attributes.message_type, MessageType::Event);
            assert_eq!(published[1].metadata.attributes.message_type, MessageType::Command);
            assert_eq!(published[2].metadata.attributes.message_type, MessageType::Query);
        }

        #[tokio::test]
        async fn test_publish_with_correlation_id() {
            let (publisher, bus) = create_simple_publisher().await;
            let correlation_id = Uuid::new_v4();
            
            let message = create_correlated_message(
                "correlated.topic",
                json!({"data": "test"}),
                correlation_id
            );
            
            publisher.publish(message).await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published[0].metadata.correlation_id, Some(correlation_id));
        }

        #[tokio::test]
        async fn test_publish_different_priorities() {
            let (publisher, bus) = create_simple_publisher().await;
            
            // 高优先级消息
            let high_priority = Message::builder("priority.topic")
                .with_payload(json!({"priority": "high"}))
                .with_priority(MessagePriority::High)
                .build();
            publisher.publish(high_priority).await.unwrap();
            
            // 普通优先级消息
            let normal_priority = Message::builder("priority.topic")
                .with_payload(json!({"priority": "normal"}))
                .with_priority(MessagePriority::Normal)
                .build();
            publisher.publish(normal_priority).await.unwrap();
            
            // 低优先级消息
            let low_priority = Message::builder("priority.topic")
                .with_payload(json!({"priority": "low"}))
                .with_priority(MessagePriority::Low)
                .build();
            publisher.publish(low_priority).await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 3);
            assert_eq!(published[0].metadata.attributes.priority, MessagePriority::High);
            assert_eq!(published[1].metadata.attributes.priority, MessagePriority::Normal);
            assert_eq!(published[2].metadata.attributes.priority, MessagePriority::Low);
        }

        #[tokio::test]
        async fn test_publish_large_message() {
            let (publisher, bus) = create_simple_publisher().await;
            
            // 创建大负载消息
            let large_data = (0..1000).map(|i| format!("item_{}", i)).collect::<Vec<_>>();
            let message = create_test_message("large.topic", json!({"data": large_data}));
            
            let result = publisher.publish(message).await;
            assert!(result.is_ok());
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
        }

        #[tokio::test]
        async fn test_publish_empty_payload() {
            let (publisher, _) = create_simple_publisher().await;
            let message = create_test_message("empty.topic", json!({}));
            
            let result = publisher.publish(message).await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_publish_null_payload() {
            let (publisher, _) = create_simple_publisher().await;
            let message = create_test_message("null.topic", json!(null));
            
            let result = publisher.publish(message).await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_publish_failure() {
            let (publisher, bus) = create_simple_publisher().await;
            bus.set_behavior("publish", MockBehavior::new().with_failure("发布失败")).await;
            
            let message = create_test_message("fail.topic", json!({"test": "data"}));
            let result = publisher.publish(message).await;
            
            assert!(result.is_err());
            assert!(result.unwrap_err().to_string().contains("发布失败"));
        }

        #[tokio::test]
        async fn test_publish_batch_empty() {
            let (publisher, bus) = create_simple_publisher().await;
            
            let result = publisher.publish_batch(vec![]).await;
            assert!(result.is_ok());
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 0);
        }

        #[tokio::test]
        async fn test_publish_batch_single_message() {
            let (publisher, bus) = create_simple_publisher().await;
            
            let message = create_test_message("batch.topic", json!({"single": true}));
            let result = publisher.publish_batch(vec![message]).await;
            
            assert!(result.is_ok());
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
        }

        #[tokio::test]
        async fn test_publish_batch_multiple_messages() {
            let (publisher, bus) = create_simple_publisher().await;
            
            let messages = create_batch_messages("batch.topic", 5);
            let result = publisher.publish_batch(messages).await;
            
            assert!(result.is_ok());
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 5);
            
            // 验证消息顺序
            for (i, msg) in published.iter().enumerate() {
                assert_eq!(msg.payload()["id"], i);
            }
        }

        #[tokio::test]
        async fn test_publish_batch_partial_failure() {
            let (publisher, bus) = create_simple_publisher().await;
            
            // 设置限制调用次数，第3次会失败
            bus.set_behavior("publish", MockBehavior::new().with_call_limit(2)).await;
            
            let messages = create_batch_messages("fail.topic", 5);
            let result = publisher.publish_batch(messages).await;
            
            assert!(result.is_err());
            
            // 前两条消息应该已经发布
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 2);
        }

        #[tokio::test]
        async fn test_publish_and_confirm() {
            let (publisher, bus) = create_simple_publisher().await;
            
            let message = create_test_message("confirm.topic", json!({"confirm": true}));
            let result = publisher.publish_and_confirm(message.clone()).await;
            
            assert!(result.is_ok());
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
            assert_eq!(published[0].payload()["confirm"], true);
        }

        #[tokio::test]
        async fn test_publish_and_confirm_failure() {
            let (publisher, bus) = create_simple_publisher().await;
            bus.set_behavior("publish", MockBehavior::new().with_failure("确认失败")).await;
            
            let message = create_test_message("fail.topic", json!({"test": "data"}));
            let result = publisher.publish_and_confirm(message).await;
            
            assert!(result.is_err());
        }

        #[tokio::test]
        async fn test_close() {
            let (publisher, _) = create_simple_publisher().await;
            
            let result = publisher.close().await;
            assert!(result.is_ok());
            
            // SimplePublisher的close方法应该总是成功
            let result2 = publisher.close().await;
            assert!(result2.is_ok());
        }
    }

    /// 测试 BatchPublisher 的所有场景
    mod batch_publisher_tests {
        use super::*;

        async fn create_batch_publisher(batch_size: usize, timeout_ms: u64) -> (BatchPublisher, Arc<MockMessageBus>) {
            let bus = Arc::new(MockMessageBus::new());
            let publisher = BatchPublisher::new(
                bus.clone(),
                "batch_publisher".to_string(),
                batch_size,
                timeout_ms
            );
            (publisher, bus)
        }

        #[tokio::test]
        async fn test_batch_publisher_creation() {
            let (publisher, _) = create_batch_publisher(10, 1000).await;
            
            assert_eq!(publisher.id(), "batch_publisher");
            assert_eq!(publisher.batch_size(), 10);
            assert_eq!(publisher.batch_timeout_ms(), 1000);
        }

        #[tokio::test]
        async fn test_batch_publish_single_message() {
            let (publisher, bus) = create_batch_publisher(10, 1000).await;
            
            let message = create_test_message("batch.topic", json!({"single": true}));
            let result = publisher.publish(message.clone()).await;
            
            assert!(result.is_ok());
            
            // 等待批处理任务处理消息
            tokio::time::sleep(Duration::from_millis(50)).await;
            
            // 手动刷新以确保消息被发送
            publisher.flush().await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
        }

        #[tokio::test]
        async fn test_batch_publish_reach_batch_size() {
            let (publisher, bus) = create_batch_publisher(3, 5000).await;
            
            // 发送3条消息，应该触发批处理
            for i in 0..3 {
                let message = create_test_message(
                    "batch.topic",
                    json!({"batch_id": i})
                );
                publisher.publish(message).await.unwrap();
            }
            
            // 等待批处理任务处理
            tokio::time::sleep(Duration::from_millis(100)).await;
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 3);
        }

        #[tokio::test]
        async fn test_batch_publish_timeout() {
            let (publisher, bus) = create_batch_publisher(10, 100).await;
            
            // 发送少于批处理大小的消息
            let message = create_test_message("timeout.topic", json!({"timeout": true}));
            publisher.publish(message).await.unwrap();
            
            // 等待超时触发批处理
            tokio::time::sleep(Duration::from_millis(150)).await;
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
        }

        #[tokio::test]
        async fn test_batch_publish_multiple_batches() {
            let (publisher, bus) = create_batch_publisher(3, 5000).await;
            
            // 发送两个完整批次的消息
            for i in 0..6 {
                let message = create_test_message(
                    "multi.batch",
                    json!({"id": i})
                );
                publisher.publish(message).await.unwrap();
            }
            
            // 等待处理
            tokio::time::sleep(Duration::from_millis(100)).await;
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 6);
        }

        #[tokio::test]
        async fn test_batch_publish_mixed_timing() {
            let (publisher, bus) = create_batch_publisher(5, 200).await;
            
            // 先发送2条消息
            for i in 0..2 {
                let message = create_test_message(
                    "mixed.topic",
                    json!({"phase": "first", "id": i})
                );
                publisher.publish(message).await.unwrap();
            }
            
            // 等待一段时间
            tokio::time::sleep(Duration::from_millis(50)).await;
            
            // 再发送3条消息，触发批处理
            for i in 2..5 {
                let message = create_test_message(
                    "mixed.topic",
                    json!({"phase": "second", "id": i})
                );
                publisher.publish(message).await.unwrap();
            }
            
            // 等待处理
            tokio::time::sleep(Duration::from_millis(100)).await;
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 5);
        }

        #[tokio::test]
        async fn test_batch_flush() {
            let (publisher, bus) = create_batch_publisher(10, 5000).await;
            
            // 发送少量消息，不会自动触发批处理
            for i in 0..3 {
                let message = create_test_message(
                    "flush.topic",
                    json!({"id": i})
                );
                publisher.publish(message).await.unwrap();
            }
            
            // 手动刷新
            let result = publisher.flush().await;
            assert!(result.is_ok());
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 3);
        }

        #[tokio::test]
        async fn test_batch_flush_empty() {
            let (publisher, _) = create_batch_publisher(10, 1000).await;
            
            // 空刷新应该成功
            let result = publisher.flush().await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_batch_publish_batch_method() {
            let (publisher, bus) = create_batch_publisher(10, 1000).await;
            
            let messages = create_batch_messages("batch.method", 5);
            let result = publisher.publish_batch(messages).await;
            
            assert!(result.is_ok());
            
            // 刷新确保所有消息都被处理
            publisher.flush().await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 5);
        }

        #[tokio::test]
        async fn test_batch_publish_and_confirm() {
            let (publisher, bus) = create_batch_publisher(5, 1000).await;
            
            let message = create_test_message("confirm.topic", json!({"confirm": true}));
            let result = publisher.publish_and_confirm(message).await;
            
            assert!(result.is_ok());
            
            publisher.flush().await.unwrap();
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
        }

        #[tokio::test]
        async fn test_batch_close() {
            let (publisher, _) = create_batch_publisher(10, 1000).await;
            
            let result = publisher.close().await;
            assert!(result.is_ok());
        }

        #[tokio::test]
        async fn test_batch_publisher_failure() {
            let (publisher, bus) = create_batch_publisher(2, 5000).await;
            
            // 设置发布失败
            bus.set_behavior("publish", MockBehavior::new().with_failure("批处理失败")).await;
            
            // 发送足够的消息触发批处理
            for i in 0..2 {
                let message = create_test_message(
                    "fail.topic",
                    json!({"id": i})
                );
                publisher.publish(message).await.unwrap();
            }
            
            // 等待处理（会失败）
            tokio::time::sleep(Duration::from_millis(100)).await;
            
            // 消息不应该被记录为已发布
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 0);
        }
    }

    /// 性能和并发测试
    mod performance_tests {
        use super::*;
        use std::sync::atomic::{AtomicUsize, Ordering};

        #[tokio::test]
        async fn test_concurrent_simple_publish() {
            let (publisher, bus) = create_simple_publisher().await;
            let publisher = Arc::new(publisher);
            let published_count = Arc::new(AtomicUsize::new(0));
            
            let mut handles = vec![];
            
            for i in 0..10 {
                let pub_clone = publisher.clone();
                let counter = published_count.clone();
                let handle = tokio::spawn(async move {
                    let message = create_test_message(
                        "concurrent.simple",
                        json!({"thread_id": i})
                    );
                    if pub_clone.publish(message).await.is_ok() {
                        counter.fetch_add(1, Ordering::Relaxed);
                    }
                });
                handles.push(handle);
            }
            
            for handle in handles {
                handle.await.unwrap();
            }
            
            assert_eq!(published_count.load(Ordering::Relaxed), 10);
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 10);
        }

        #[tokio::test]
        async fn test_concurrent_batch_publish() {
            let (publisher, bus) = create_batch_publisher(5, 1000).await;
            let publisher = Arc::new(publisher);
            let published_count = Arc::new(AtomicUsize::new(0));
            
            let mut handles = vec![];
            
            for i in 0..20 {
                let pub_clone = publisher.clone();
                let counter = published_count.clone();
                let handle = tokio::spawn(async move {
                    let message = create_test_message(
                        "concurrent.batch",
                        json!({"thread_id": i})
                    );
                    if pub_clone.publish(message).await.is_ok() {
                        counter.fetch_add(1, Ordering::Relaxed);
                    }
                });
                handles.push(handle);
            }
            
            for handle in handles {
                handle.await.unwrap();
            }
            
            // 确保所有消息都被处理
            publisher.flush().await.unwrap();
            
            assert_eq!(published_count.load(Ordering::Relaxed), 20);
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 20);
        }

        #[tokio::test]
        async fn test_publish_throughput() {
            let (publisher, bus) = create_simple_publisher().await;
            
            let message_count = 1000;
            let start_time = std::time::Instant::now();
            
            for i in 0..message_count {
                let message = create_test_message(
                    "throughput.test",
                    json!({"id": i})
                );
                publisher.publish(message).await.unwrap();
            }
            
            let duration = start_time.elapsed();
            let throughput = message_count as f64 / duration.as_secs_f64();
            
            println!("发布吞吐量: {:.2} 消息/秒", throughput);
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), message_count);
        }

        #[tokio::test]
        async fn test_batch_throughput() {
            let (publisher, bus) = create_batch_publisher(100, 1000).await;
            
            let message_count = 1000;
            let start_time = std::time::Instant::now();
            
            for i in 0..message_count {
                let message = create_test_message(
                    "batch.throughput",
                    json!({"id": i})
                );
                publisher.publish(message).await.unwrap();
            }
            
            // 等待所有批次处理完成
            publisher.flush().await.unwrap();
            
            let duration = start_time.elapsed();
            let throughput = message_count as f64 / duration.as_secs_f64();
            
            println!("批处理吞吐量: {:.2} 消息/秒", throughput);
            
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), message_count);
        }
    }

    /// 集成测试：Publisher与其他组件的交互
    mod integration_tests {
        use super::*;

        #[tokio::test]
        async fn test_publisher_subscriber_integration() {
            let bus = Arc::new(MockMessageBus::new());
            let publisher = SimplePublisher::new(bus.clone(), "integration_pub".to_string());
            
            // 创建订阅者
            let subscriber = bus.subscribe("integration.test").await.unwrap();
            
            // 发布消息
            let message = create_test_message(
                "integration.test",
                json!({"integration": "test"})
            );
            publisher.publish(message.clone()).await.unwrap();
            
            // 验证消息被正确路由
            let published = bus.get_published_messages_for_topic("integration.test").await;
            assert_eq!(published.len(), 1);
            assert_eq!(published[0].payload()["integration"], "test");
        }

        #[tokio::test]
        async fn test_multiple_publishers_same_topic() {
            let bus = Arc::new(MockMessageBus::new());
            let pub1 = SimplePublisher::new(bus.clone(), "publisher_1".to_string());
            let pub2 = SimplePublisher::new(bus.clone(), "publisher_2".to_string());
            
            // 两个发布者发送到同一主题
            let msg1 = create_test_message("shared.topic", json!({"from": "pub1"}));
            let msg2 = create_test_message("shared.topic", json!({"from": "pub2"}));
            
            pub1.publish(msg1).await.unwrap();
            pub2.publish(msg2).await.unwrap();
            
            let published = bus.get_published_messages_for_topic("shared.topic").await;
            assert_eq!(published.len(), 2);
            
            let sources: Vec<&str> = published.iter()
                .map(|m| m.payload()["from"].as_str().unwrap())
                .collect();
            assert!(sources.contains(&"pub1"));
            assert!(sources.contains(&"pub2"));
        }

        #[tokio::test]
        async fn test_publisher_error_handling_flow() {
            let bus = Arc::new(MockMessageBus::new());
            let publisher = SimplePublisher::new(bus.clone(), "error_test_pub".to_string());
            
            // 设置第2次发布失败
            bus.set_behavior("publish", MockBehavior::new().with_call_limit(1)).await;
            
            // 第一次发布应该成功
            let msg1 = create_test_message("error.test", json!({"attempt": 1}));
            let result1 = publisher.publish(msg1).await;
            assert!(result1.is_ok());
            
            // 第二次发布应该失败
            let msg2 = create_test_message("error.test", json!({"attempt": 2}));
            let result2 = publisher.publish(msg2).await;
            assert!(result2.is_err());
            
            // 验证只有第一条消息被发布
            let published = bus.get_published_messages().await;
            assert_eq!(published.len(), 1);
            assert_eq!(published[0].payload()["attempt"], 1);
        }

        #[tokio::test]
        async fn test_mixed_publisher_types() {
            let bus = Arc::new(MockMessageBus::new());
            let simple_pub = SimplePublisher::new(bus.clone(), "simple".to_string());
            let batch_pub = BatchPublisher::new(bus.clone(), "batch".to_string(), 3, 1000);
            
            // 简单发布者发送消息
            let simple_msg = create_test_message("mixed.test", json!({"type": "simple"}));
            simple_pub.publish(simple_msg).await.unwrap();
            
            // 批处理发布者发送消息
            let batch_msg = create_test_message("mixed.test", json!({"type": "batch"}));
            batch_pub.publish(batch_msg).await.unwrap();
            batch_pub.flush().await.unwrap();
            
            let published = bus.get_published_messages_for_topic("mixed.test").await;
            assert_eq!(published.len(), 2);
            
            let types: Vec<&str> = published.iter()
                .map(|m| m.payload()["type"].as_str().unwrap())
                .collect();
            assert!(types.contains(&"simple"));
            assert!(types.contains(&"batch"));
        }
    }
}